package com.wish.email.consumer;

import com.alibaba.fastjson.JSON;
import com.wish.dto.contract.SaveContractDto;
import com.wish.email.service.MailService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class KafkaConsumer implements AcknowledgingMessageListener<String,String> {

    @Autowired
    MailService mailService;

    @Autowired
    RedisTemplate redisTemplate;

    @Override
    @KafkaListener(topicPartitions = {@TopicPartition(topic = "contract",partitions = "0")})
    public void onMessage(ConsumerRecord<String, String> consumerRecord, Acknowledgment acknowledgment) {

         // key唯一，防止重复消费
        String key = consumerRecord.key();
        if(!redisTemplate.opsForValue().setIfAbsent(key,null,1,TimeUnit.DAYS)){
            log.info("kafka防止重复消费");
            return;
        }
        // 获取邮件内容数据
        String value = consumerRecord.value();
        SaveContractDto dto = JSON.parseObject(value, SaveContractDto.class);
        String context = "合同编号：%s,租房人：%s,出租人： %s,合同期限： %s";
        String mailData = String.format(context, dto.getContractNo(), dto.getRenterId(), dto.getLandlordId(), dto.getStartDate());
        //发送邮件
        mailService.sendEmail("合同编号"+dto.getContractNo(),"894910150@qq.com","1134895347@qq.com",mailData);
        //手动提交
        acknowledgment.acknowledge();
        }
    }
